package com.edu.elasticsearch.common;

import java.io.IOException;
import java.math.BigDecimal;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.log4j.Logger;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import com.edu.elasticsearch.beans.ESQueryBean;
import com.edu.elasticsearch.utils.ElasticSearchUtils;

public class ElasticSearchService {
	
	private static Logger logger = Logger.getLogger(ElasticSearchService.class);
	
	private static TransportClient client;
	private static String cluster_name = "demo-application";
	private static String host_name="127.0.0.1";
	private static Integer host_port = 9300;
	private static String index_name = "xcsm2";
	private static String index_type = "shop_info2";
	
	@SuppressWarnings("unchecked")
	public static TransportClient getClient(){
		try {
			if(null == client){
				Settings settings = Settings.builder().put("cluster.name", cluster_name).build();
				client = new PreBuiltTransportClient(settings);
				client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host_name), host_port));
			}
			return client;
		} catch (UnknownHostException e) {
			logger.error("初始化elasticsearch失败", e);
		}
		return null;
	}
	
	// 创建索引 (类似数据库的建表语句)
    public static void createIndex(){ 
    	try {
			HashMap<String, Object> settings_map = new HashMap<String, Object>(2);
			settings_map.put("number_of_shards", 5);
			settings_map.put("number_of_replicas", 1);
			// 创建Mapping  
			XContentBuilder mapping = ElasticSearchUtils.createMapping(index_type);  
			logger.info("mapping:" + mapping.string());  
			// 创建一个空索引  
			getClient().admin().indices().prepareCreate(index_name).setSettings(settings_map).execute().actionGet();  
			PutMappingRequest putMapping = Requests.putMappingRequest(index_name).type(index_type).source(mapping);  
			PutMappingResponse response = getClient().admin().indices().putMapping(putMapping).actionGet();  
			if (!response.isAcknowledged()) {  
				logger.error("Could not define mapping for type [" + index_name + "]/[" + index_type + "].");  
			} else {  
				logger.info("Mapping definition for [" + index_name + "]/[" + index_type + "] succesfully created.");  
			}
		} catch (IOException e) {
			logger.error("创建索引错误！",e); 
		}  
    } 
    
    // 批量创建索引  
	public static int insert(List<Map<String, Object>> list){
        BulkRequestBuilder bulkRequest = getClient().prepareBulk();  
        for (Map<String,Object> map : list) {  
        	IndexRequest request = getClient().prepareIndex(index_name, index_type, String.valueOf(map.get("id"))).setSource(map).request();
            bulkRequest.add(request);  
        }  
        BulkResponse bulkResponse = bulkRequest.execute().actionGet();  
        if (bulkResponse.hasFailures()) {  
        	logger.error("批量创建索引错误！"+bulkResponse.buildFailureMessage());  
        }  
        return bulkRequest.numberOfActions();  
	}
	
	/**
	* @Description: 更新 
	* @param id 
	* @param storeData
	 */
	public static void update(String id,Map<String,String> storeData){
		try {
			XContentBuilder jsonData = XContentFactory.jsonBuilder();
			jsonData.startObject();
			for (Map.Entry<String,String> entry : storeData.entrySet()) { 
				jsonData.field(entry.getKey(), entry.getValue());
			}
			jsonData.endObject();
			IndexRequest indexRequest = new IndexRequest(index_name,index_type, id).source(jsonData);
			UpdateRequest updateRequest = new UpdateRequest(index_name,index_type, id).doc(jsonData);
			updateRequest.upsert(indexRequest);
			getClient().update(updateRequest);
		} catch (Exception e) {
			logger.error("更新失败", e);
		}
	}
	
	//删除
	public static String delete(String id){
		DeleteResponse response = getClient().prepareDelete(index_name,index_type, id).get();
		return response.getId();
	}
	
	public static List<Map<String,Object>> query(QueryBuilder queryBuilder, ESQueryBean queryBean){
		List<Map<String,Object>> list = new ArrayList<Map<String,Object>>();
		Long startTime = System.currentTimeMillis();
		SearchResponse searchResponse = client.prepareSearch(index_name)
				.setTypes(index_type)
				.setQuery(queryBuilder)
				.addSort(queryBean.getSortBuilder())
				.setFrom(queryBean.getPage()).setSize(queryBean.getSize())
				.execute().actionGet();
		Long endTime = System.currentTimeMillis();
		SearchHits hits = searchResponse.getHits();
		logger.debug("查询到记录数=" + hits.getTotalHits());
		SearchHit[] searchHists = hits.getHits();
		if (searchHists.length > 0) {
			for (SearchHit hit : searchHists) {
				Map<String, Object> hitMap = hit.getSource();  
				Object[] sortValues = hit.getSortValues();
				if(null == sortValues || sortValues.length>0){
					String  sort = sortValues[0].toString();
					if(!StringUtils.isEmpty(sort) && NumberUtils.isNumber(sort)){
						// 获取距离值，并保留两位小数点  
						BigDecimal geoDis = new BigDecimal(sort);  
						hitMap.put("distance", geoDis.setScale(2, BigDecimal.ROUND_HALF_DOWN));  
					}
				}
				list.add(hitMap);
			}
		}
		logger.debug("--elasticsearch--endTime-startTime="+(endTime-startTime));
		return list;
	}
}
